package com.ada.flink

import com.ada.flink.bean.StartUpLog
import com.ada.flink.util.MyKafkaUtil
import com.alibaba.fastjson.JSON
import org.apache.flink.api.java.tuple.Tuple
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.{DataStream, KeyedStream, StreamExecutionEnvironment, WindowedStream}
import org.apache.flink.streaming.api.windowing.assigners.{EventTimeSessionWindows, SlidingEventTimeWindows, TumblingEventTimeWindows}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow

import scala.collection.mutable

object WindowApiApp {
    def main(args: Array[String]): Unit = {

        val environment: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

        val kafkaConsumer = MyKafkaUtil.getConsumer("GMALL_STARTUP")

        import org.apache.flink.api.scala._

        val dstream: DataStream[String] = environment.addSource(kafkaConsumer)

        //dstream.print()
        //求各个渠道的累计个数
        val startUplogDstream: DataStream[StartUpLog] = dstream.map {
            JSON.parseObject(_, classOf[StartUpLog])
        }

        val textWithTsDstream: DataStream[(String, Long, Int)] = dstream.map { text =>
            val arr: Array[String] = text.split(" ")
            (arr(0), arr(1).toLong, 1)
        }

        /*
        滚动窗口
        Flink默认的时间窗口根据ProcessingTime进行窗口的划分，将Flink获取到的数据根据进入Flink的时间划分到不同的窗口中。
        时间间隔可以通过Time.milliseconds(x)，Time.seconds(x)，Time.minutes(x)等其中的一个来指定。
         */
        val keyedStream1: KeyedStream[(String, Int), Tuple] = startUplogDstream.map(startuplog=>(startuplog.ch,1)).keyBy(0)
        //每10统计一次各个渠道的计数
        val windowedStream1: WindowedStream[(String, Int), Tuple, TimeWindow] = keyedStream1.timeWindow(Time.seconds(10))
        val sumDstream1: DataStream[(String, Int)] = windowedStream1.sum(1)

        /*
        滑动窗口（SlidingEventTimeWindows）
        滑动窗口和滚动窗口的函数名是完全一致的，只是在传参数时需要传入两个参数，一个是window_size，一个是sliding_size。
        下面代码中的sliding_size设置为了2s，也就是说，窗口每2s就计算一次，每一次计算的window范围是5s内的所有元素。
        时间间隔可以通过Time.milliseconds(x)，Time.seconds(x)，Time.minutes(x)等其中的一个来指定。
         */
        val keyedStream2: KeyedStream[(String, Int), Tuple] = startUplogDstream.map(startuplog=>(startuplog.ch,1)).keyBy(0)
        //每5秒统计一次最近10秒的各个渠道的计数
        val windowedStream2: WindowedStream[(String, Int), Tuple, TimeWindow] = keyedStream2.timeWindow(Time.seconds(10),Time.seconds(5))
        val sumDstream2: DataStream[(String, Int)] = windowedStream2.sum(1)

        /*
        EventTime的引入
        在Flink的流式处理中，绝大部分的业务都会使用eventTime，一般只在eventTime无法使用时，才会被迫使用ProcessingTime或者IngestionTime。
        如果要使用EventTime，那么需要引入EventTime的时间属性，引入方式如下所示：
         */
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        // 从调用时刻开始给env创建的每一个stream追加时间特征
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

        /*
        Watermark的引入
        Watermark是一种衡量Event Time进展的机制，它是数据本身的一个隐藏属性，数据本身携带着对应的Watermark。
        Watermark是用于处理乱序事件的，而正确的处理乱序事件，通常用Watermark机制结合window来实现。
        数据流中的Watermark用于表示timestamp小于Watermark的数据，都已经到达了，因此，window的执行也是由Watermark触发的。
        Watermark可以理解成一个延迟触发机制，我们可以设置Watermark的延时时长t，每次系统会校验已经到达的数据中最大的maxEventTime，然后认定eventTime小于maxEventTime- t的所有数据都已经到达，如果有窗口的停止时间等于maxEventTime – t，那么这个窗口被触发执行。
        Watermark 就是触发前一窗口的“关窗时间”，一旦触发关门那么以当前时刻为准在窗口范围内的所有所有数据都会收入窗中。
        只要没有达到水位那么不管现实中的时间推进了多久都不会触发关窗。
         */
        val textWithEventTimeDstream: DataStream[(String, Long, Int)] = textWithTsDstream.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[(String, Long, Int)](Time.milliseconds(1000)) {
            override def extractTimestamp(element: (String, Long, Int)): Long = {
                return  element._2
            }
        })


        val textKeyStream: KeyedStream[(String, Long, Int), Tuple] = textWithEventTimeDstream.keyBy(0)
        textKeyStream.print("textkey:")

        /*
        滚动窗口（TumblingEventTimeWindows）
        结果是按照Event Time的时间窗口计算得出的，而无关系统的时间（包括输入的快慢）。
         */
        val windowStream: WindowedStream[(String, Long, Int), Tuple, TimeWindow] = textKeyStream.window(TumblingEventTimeWindows.of(Time.seconds(2)))
        val groupDstream: DataStream[mutable.HashSet[Long]] = windowStream.fold(new mutable.HashSet[Long]()) { case (set, (key, ts, count)) =>
            set += ts
        }
        groupDstream.print("window::::").setParallelism(1)

        /*
        滑动窗口（SlidingEventTimeWindows）
         */
        val windowStream2: WindowedStream[(String, Long, Int), Tuple, TimeWindow] = textKeyStream.window(SlidingEventTimeWindows.of(Time.seconds(2),Time.milliseconds(500)))
        val groupDstream2: DataStream[mutable.HashSet[Long]] = windowStream.fold(new mutable.HashSet[Long]()) { case (set, (key, ts, count)) =>
            set += ts
        }
        groupDstream2.print("window::::").setParallelism(1)

        /*
        会话窗口（EventTimeSessionWindows）
        相邻两次数据的EventTime的时间差超过指定的时间间隔就会触发执行。如果加入Watermark，会在符合窗口触发的情况下进行延迟。到达延迟水位再进行窗口触发。
         */
        val windowStream3: WindowedStream[(String, Long, Int), Tuple, TimeWindow] = textKeyStream.window(EventTimeSessionWindows.withGap(Time.milliseconds(500)) )
        windowStream3.reduce((text1,text2)=>
            (  text1._1,0L,text1._3+text2._3)
        )  .map(_._3).print("windows:::").setParallelism(1)


        env.execute()
    }
}
